package net.lmc.bwie.realtime.jtp.common.utils;

import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.HashMap;
import java.util.Map;

public class HbaseUtil {

    private static Connection connection = null;

    static {
        try {
            // 1.创建配置对象，设置属性值，比如连接ZK集群信息
            org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
            conf.set(HConstants.CLIENT_ZOOKEEPER_QUORUM, "node101,node102,node103");
            conf.set(HConstants.CLIENT_ZOOKEEPER_CLIENT_PORT, "2181");
            // 2.实例化Connection连接
            connection = ConnectionFactory.createConnection(conf);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 依据RowKey获取HBase表中数据
     *
     * @param tableName  表名称
     * @param rowKey     主键
     * @param familyName 列簇名称
     * @param columnName 列名称
     */
    public static String getData(String tableName, String rowKey,
                                 String familyName, String columnName) throws Exception {
        // a.实例化Table对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        // b. 获取数据
        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = table.get(get);
        // c. 获取字段值
        byte[] resultValue = result.getValue(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
        table.close();
        // d. 返回
        return Bytes.toString(resultValue);
    }

    /**
     * 扫描HBase表数据
     *
     * @param tableName 表名称
     * @param family    列簇名称
     */
    public static Map<String, String> scanData(String tableName, String family, String column) throws Exception {
        // a.实例化Table对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        // b.构建扫描器对象
        Scan scan = new Scan();
        scan.addFamily(Bytes.toBytes(family));
        // c. 扫描数据
        ResultScanner scanner = table.getScanner(scan);
        // d. 提取数据
        Map<String, String> dimMap = new HashMap<>();
        for (Result result : scanner) {
            // RowKey行健
            String rowKey = Bytes.toString(result.getRow());
            // 列值
            String columnValue = Bytes.toString(result.getValue(Bytes.toBytes(family), Bytes.toBytes(column)));
            // 放入集合
            dimMap.put(rowKey, columnValue);
        }
        // e. 关闭
        scanner.close();
        table.close();
        // f. 返回对象
        return dimMap;
    }

    /**
     * 关闭链接
     */
    public static void close() throws Exception {
        if (null != connection) {
            connection.close();
        }
    }
}

